from kafka import KafkaConsumer
import json
import time

# ==================== 配置区 ====================
KAFKA_BOOTSTRAP_SERVERS = ['192.168.162.118:9092']
TOPIC_NAME = 'LAB-HTTP'
SEARCH_KEYWORD = 'cyber'
MAX_RECORDS = 40  # 只打印前 40 条
# ================================================

print(f"🔍 正在从 Kafka 主题 '{TOPIC_NAME}' 读取数据，仅提取前 {MAX_RECORDS} 条包含 '{SEARCH_KEYWORD}' 的 DPI 攻击记录...\n")
print(f"{'序号':<3} {'sid':<15} {'spt':<6} {'dip':<15} {'dpt':<6} {'cid':<32} {'时间':<19} {'方法':<8} {'ACRM'}")
print("-" * 120)

consumer = None
count = 0

try:
    consumer = KafkaConsumer(
        TOPIC_NAME,
        bootstrap_servers=KAFKA_BOOTSTRAP_SERVERS,
        auto_offset_reset='earliest',
        enable_auto_commit=False,
        group_id='cyber-printer-8field',
        value_deserializer=lambda x: x,
        consumer_timeout_ms=10000
    )

    for message in consumer:
        if count >= MAX_RECORDS:
            print("\n✅ 已打印 40 条记录，停止。")
            break

        raw_bytes = message.value

        # 解压 GZIP（如果存在）
        if len(raw_bytes) >= 2 and raw_bytes[:2] == b'\x1f\x8b':
            try:
                raw_bytes = gzip.decompress(raw_bytes)
            except Exception:
                continue

        # 解码为 UTF-8
        try:
            full_content = raw_bytes.decode('utf-8')
        except UnicodeDecodeError:
            continue

        # ✅ 只处理以 "Header:" 开头的 DPI 流量
        if not full_content.strip().startswith("Header:"):
            continue

        # ✅ 检查是否包含 'cyber'（不区分大小写）
        if SEARCH_KEYWORD.lower() not in full_content.lower():
            continue

        # ✅ 解析 Header JSON（提取 sid, spt, dip, dpt, cid, et）
        lines = full_content.splitlines()
        header_line = lines[0]

        try:
            header_json = json.loads(header_line[7:])  # 去掉 "Header:"
            sid = header_json.get("sip", "UNKNOWN")
            spt = header_json.get("spt", "UNKNOWN")
            dip = header_json.get("dip", "UNKNOWN")
            dpt = header_json.get("dpt", "UNKNOWN")
            cid = header_json.get("cid", "UNKNOWN")
            et_timestamp = header_json.get("et", 0)
            et_human = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(et_timestamp / 1000)) if et_timestamp else "UNKNOWN"
        except Exception:
            continue

        # ✅ 提取 HTTP 方法 和 ACRM
        http_method = "UNKNOWN"
        acrm = "N/A"

        for line in lines[1:]:
            if line.startswith(("OPTIONS", "GET", "POST", "PUT", "DELETE", "HEAD", "PATCH")):
                http_method = line.split()[0]
            elif line.startswith("Access-Control-Request-Method:"):
                acrm = line.split(":", 1)[1].strip()

        # ✅ 打印一行，格式对齐（使用制表符和空格对齐）
        count += 1
        print(f"{count:<3} {sid:<15} {spt:<6} {dip:<15} {dpt:<6} {cid:<32} {et_human:<19} {http_method:<8} {acrm}")

    if count == 0:
        print("\n❌ 未找到任何符合条件的记录（Header: + cyber）")
    else:
        print(f"\n✅ 已成功提取 {count} 条攻击记录。")

except KeyboardInterrupt:
    print("\n🛑 用户中断。")
except Exception as e:
    print(f"\n⚠️ 发生异常: {e}")
finally:
    if consumer:
        consumer.close()
    print("🔌 消费者已关闭。")
